#include "pthreadpool.h"
#include "handleEvent.h"

pThreadPool::pThreadPool()
{
  min_socknum_idx_ = 0;
  cur_idx_ = 0;
}

pThreadPool::~pThreadPool()
{
  for(size_t i = 0; i < thread_epollfd_vec_.size(); ++i)
  {
    close(thread_epollfd_vec_[i]);
  }
}


bool pThreadPool::init(uint32 threadnum)
{
  thread_epollfd_vec_.resize(threadnum);
	for (uint32 i = 0; i < threadnum; ++i)
	{
		int epollfd = EpollOpt::CreateEpoll();
		if (epollfd <= 0)
		{
			return false;
		}
    std::cout << "-------" << epollfd << std::endl;
    //可能扩容
		thread_epollfd_vec_[i] = epollfd;
    idx_socknum_map_[i] = 0;

		pthread_t tid;
		if (0 != pthread_create(&tid, nullptr, run, &thread_epollfd_vec_[i]))
		{
			LOG_ERROR("pthread init create thread fail");
			return false;
		}
		pthread_detach(tid);
	}
	return true;
}

#define EPOLLEVENT_NUM 8
void* pThreadPool::run(void* arg)
{
	int epollfd = *(int*)arg;
	if (epollfd <= 0)
  { 
    LOG_ERROR("FFL[%s:%s:%u], epollfd:%d",LOG_FFL, epollfd);
		return nullptr;
  }
  LOG_INFO("FFL[%s:%s:%u] run thread threadid:%u epollfd:%d", LOG_FFL, pthread_self(), epollfd);
	while (true)
	{
		struct epoll_event evts[EPOLLEVENT_NUM];
		int num = EpollOpt::WaitEpoll(epollfd, evts, EPOLLEVENT_NUM, -1);
		if (num == 0)
		{
			LOG_INFO("FFL[%s:%s:%u] wait epoll timeout", LOG_FFL);
		}
		else if (num < 0)
		{
			LOG_ERROR("FFL[%s:%s:%u] wait epoll fail, epollfd:%d", LOG_FFL, epollfd);
			return nullptr;
		}
		else
		{
			for (int i = 0; i < num; ++i)
			{
				//处理recv和send
				struct epoll_event ev = evts[i];

				task* tk = (task*)ev.data.ptr;
				if (tk)
				{
					HANDLE func = (HANDLE)tk->et_;
					func(tk);
				}
			}
		}
	}
}

bool pThreadPool::addThreadEpoll(int clientfd)
{
	if (thread_epollfd_vec_.empty())
		return false;

	//int idx = clientfd % thread_epollfd_vec_.size();
	int epollfd = thread_epollfd_vec_[cur_idx_];
  if(epollfd < 0)
  {
	  while(epollfd < 0)
    {
      cur_idx_++;
      if(cur_idx_ >= thread_epollfd_vec_.size())
      {
        return false;
      }
      cur_idx_ %= thread_epollfd_vec_.size();
      epollfd = thread_epollfd_vec_[cur_idx_];
    }
  }
  else 
  {
    cur_idx_++;
    cur_idx_ %= thread_epollfd_vec_.size();
  }

	task* tk = new task;
	tk->epollfd_ = epollfd;
	tk->fd_ = clientfd;
	tk->et_ = (void*)HandleEvent::handleOpensslRecv;
	EpollOpt::AddEpoll(epollfd, clientfd, EPOLLIN, (void*)tk);
  return true;
}
	
bool pThreadPool::addThreadEpollNew(int clientfd)
{
  if(thread_epollfd_vec_.empty() || idx_socknum_map_.empty())
    return false;;

  uint32 idx = getMinSockNumIdx();
  int epollfd = thread_epollfd_vec_[idx];
  if(epollfd < 0)
  {
    LOG_ERROR("FFL[%s:%s:%u] thread epollfd error fd:%d", LOG_FFL, epollfd);
    return false;
  }

  addEpollSockNum(idx);

  task* tk = new task;
  tk->idx_ = idx;
	tk->epollfd_ = epollfd;
	tk->fd_ = clientfd;
	tk->et_ = (void*)HandleEvent::handleOpensslRecv;
	EpollOpt::AddEpoll(epollfd, clientfd, EPOLLIN, (void*)tk);
  return true;
}
  

int pThreadPool::getThreadEpollfd(int& index)
{
  if(thread_epollfd_vec_.empty() || idx_socknum_map_.empty())
    return -1;

  uint32 idx = getMinSockNumIdx();
  int epollfd = thread_epollfd_vec_[idx];
  if(epollfd < 0)
  {
    LOG_ERROR("FFL[%s:%s:%u] thread epollfd error fd:%d", LOG_FFL, epollfd);
    return -1;
  }
  index = idx;
  return epollfd;
}

void pThreadPool::addEpollSockNum(uint32 idx)
{
  mt_.lock();
  if(idx >= thread_epollfd_vec_.size())
  {
    mt_.unlock();
    return;
  }

  idx_socknum_map_[idx]++;

  uint32 max_num = UINT_MAX;
  std::map<uint32, uint32>::const_iterator cit = idx_socknum_map_.begin();
  for(; cit != idx_socknum_map_.end(); ++cit)
  {
    if(cit->second < max_num)
    {
      max_num = cit->second;
      min_socknum_idx_ = cit->first;
    }
  }
  mt_.unlock();
}
  
void pThreadPool::subEpollSockNum(uint32 idx)
{
  mt_.lock();
  if(idx >= thread_epollfd_vec_.size())
  {
    mt_.unlock();
    return;
  }
  uint32& sock_num = idx_socknum_map_[idx];
  if(sock_num <= 0)
  {
    mt_.unlock();
    return;
  }
  sock_num--;

  uint32 max_num = UINT_MAX;
  std::map<uint32, uint32>::const_iterator cit = idx_socknum_map_.begin();
  for(; cit != idx_socknum_map_.end(); ++cit)
  {
    if(cit->second < max_num)
    {
      max_num = cit->second;
      min_socknum_idx_ = cit->first;
    }
  }
  mt_.unlock();
}

uint32 pThreadPool::getMinSockNumIdx()
{
  return min_socknum_idx_;
}
